package com.simple.datax.conn.impl;

import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.Map;

import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.hornetq.api.core.TransportConfiguration;
import org.hornetq.api.jms.HornetQJMSClient;
import org.hornetq.api.jms.JMSFactoryType;
import org.hornetq.core.remoting.impl.netty.NettyConnectorFactory;
import org.hornetq.core.remoting.impl.netty.TransportConstants;
import org.hornetq.jms.client.HornetQConnectionFactory;

import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.simple.datax.SimpleMessage;
import com.simple.datax.api.ConnectorException;

/** 
 * @Title: HornetqConnImpl.java 
 * @Package com.simple.datax.conn.impl 
 * @Description: 实现具体的HornetQ连接操作，包括连接，接收，发送等。 
 * @author Jin Yi 
 * @date 2012-12-30
 * @version V1.0 
 */
public class HornetqConnImpl extends MQBaseConnector implements MessageListener, ExceptionListener {

	private static final Logger logger = LoggerFactory
			.getLogger(HornetqConnImpl.class);
	public static final String TYPE = "HornetQ";
	private HornetQConnectionFactory cf;
	private String mqName;
	private Queue queue;
	private Connection conn;
	private String jmsUrl = "";
	private int jmsPort; 
	private Session session = null;
	private MessageProducer producer = null;
	private MessageConsumer consumer = null;
	
	/**
	 * 构造器
	 * @param connPara
	 * @throws ConnectorException
	 */
	public HornetqConnImpl(String connPara) throws ConnectorException {
		super(connPara);
		setConnPara(connPara);
		parseConnPara();
		init();
	}
	
	/**
	 * HornetQ连接器初始化
	 * @throws ConnectorException
	 */
	public void init() throws ConnectorException {
		Map<String, Object> connectionParams = new HashMap<String, Object>();
		connectionParams.put(TransportConstants.PORT_PROP_NAME, jmsPort);
		connectionParams.put(TransportConstants.HOST_PROP_NAME, jmsUrl);
		TransportConfiguration transportConfiguration = new TransportConfiguration(
				NettyConnectorFactory.class.getName(), connectionParams);
		cf = (HornetQConnectionFactory) HornetQJMSClient
				.createConnectionFactoryWithoutHA(JMSFactoryType.CF,
						transportConfiguration);
		cf.setRetryInterval(2000);
		cf.setRetryIntervalMultiplier(1.5);
		cf.setMaxRetryInterval(20000);
		cf.setReconnectAttempts(-1);
		cf.setClientFailureCheckPeriod(10000);
		try {
			conn = cf.createConnection();
			session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
			conn.start();
			conn.setExceptionListener(this);
		} catch (JMSException e) {
			throw new ConnectorException("Initialized Hornetq failed.", e);
		}
	}
	
	@Override
	public String getMessage() throws ConnectorException {
		String ret = null;
		prepareReceiveQueue();
		try {
			Message message = consumer.receive(RECV_NO_MESSAGE_TIMOUT);
			
			if (message instanceof TextMessage) {
				TextMessage textMsg = (TextMessage) message;
				if (textMsg != null) {
					ret = textMsg.getText();
					ret = getInputString(ret, getEncoding());
				}
			}
			if (message instanceof BytesMessage) {
				BytesMessage bytesMsg = (BytesMessage) message;
				if (bytesMsg != null) {
					ret = getMessage(bytesMsg, getEncoding());
				}
			}
			
		} catch (JMSException ex) {
			throw new ConnectorException("Receive message from Hornetq failed.", ex);
		} catch (UnsupportedEncodingException e) {
			throw new ConnectorException("Receive message from Hornetq failed.", e);
		}
		return ret;
	}

	@Override
	protected void prepareSend() throws ConnectorException {
		if (conn == null || queue == null || consumer == null) {
			try {
				queue = null;
				queue = session.createQueue(mqName);
				producer = null;
				producer = session.createProducer(queue);
			} catch (Exception ex) {
				throw new ConnectorException("Hornetq init failed.", ex);
			}
		}
	}

	@Override
	protected void internalSend(SimpleMessage message)
			throws ConnectorException {
//		TextMessage textMsg = null;
		try {
//			textMsg = session.createTextMessage();
//			String content = getOutputString(message.getContent(), getEncoding());
//			textMsg.setText(content);
			
			BytesMessage msg = session.createBytesMessage();
			msg.writeBytes(getOutputBytes(message.getContent(), getEncoding()));
			producer.send(msg);
			logger.debug("Send out message to Hornetq. ");
		} catch (JMSException e) {
			throw new ConnectorException("Sent message failed to Hornetq.", e, true);
		} catch (Exception e) {
			throw new ConnectorException("Sent message failed to Hornetq.", e);
		}

	}

	@Override
	public String getConnType() {
		return TYPE;
	}
	
	/**
	 * 解析连接参数
	 * @throws ConnectorException 
	 */
	private void parseConnPara() throws ConnectorException {
		//connPara = "IP:PORT:USER:PASSWORD:PATH";
		String[] tmpStr = splitConnPara(connPara);
		if (tmpStr.length==3) {
			jmsPort = Integer.parseInt(tmpStr[1]);
			jmsUrl = tmpStr[0];
			mqName = tmpStr[2];
		} else {
			throw new ConnectorException(
					"Parsing Hornetq Connector parameters failed. Please check parameter - "
							+ connPara);
		}
	}

	@Override
	public void onException(JMSException ex) {
		logger.error("HornetQ got Exception.", ex);
	}
	
	/**
	 * 为接收消息做准备
	 * @throws ConnectorException
	 */
	private void prepareReceiveQueue() throws ConnectorException{
		try {
			if (conn == null || queue == null || consumer == null) {
				queue = null;
				queue = session.createQueue(mqName);
				consumer = null;
				consumer = session.createConsumer(queue);
			}
		} catch (JMSException e) {
			throw new ConnectorException("Hornetq producer initialization failed", e);
		}
	}
	
	public static void main(String[] args) {
		HornetqConnImpl conn = null;
		try {
			conn = new HornetqConnImpl("localhost$5445$DLQ");
			String str = conn.getMessage();
			System.out.println(str);
		} catch (ConnectorException e) {
			e.printStackTrace();
		} finally {
			conn.disconnect();
		}
	}

	@Override
	public void disconnect() {
		if (producer != null) {
			try {
				producer.close();
			} catch (JMSException e) {
				logger.warn("HornetQ producer disconnect error", e);
			}
			producer = null;
		}
		
		if (consumer != null) {
			try {
				consumer.close();
			} catch (JMSException e) {
				logger.warn("HornetQ consumer disconnect error", e);
			}
			producer = null;
		}
		
		if (session != null) {
			try {
				session.close();
			} catch (JMSException e) {
				logger.warn("HornetQ session disconnect error", e);
			}
			session = null;
		}
		
		if (conn != null) {
			try {
				conn.close();
			} catch (JMSException e) {
				logger.warn("HornetQ connection disconnect error", e);
			}
			conn = null;
		}
	}

	@Override
	public void onMessage(Message message) {
		// 目前不使用监听接口来接收消息
//		if (message instanceof TextMessage) {
//			TextMessage textMsg = (TextMessage) message;
//		} else {
//			logger.error("Get invalid message type "+ message.getClass().getName() +" from HornetQ. ");
//		}
	}

	@Override
	protected void prepareReceive() throws ConnectorException {
		prepareReceiveQueue();
	}

	@Override
	public void setMessageListener(MessageListener listener)
			throws ConnectorException {
		// TODO Auto-generated method stub
		
	}
}
